package org.yaac.server.egql;

import static com.google.common.collect.Lists.newArrayList;
import static com.google.common.collect.Sets.newHashSet;

import java.util.Collection;
import java.util.List;
import java.util.Set;

import org.yaac.server.egql.evaluator.AggregationEvaluator;
import org.yaac.server.egql.evaluator.Evaluator;
import org.yaac.server.egql.exception.EGQLE001Exception;
import org.yaac.server.egql.exception.EGQLE002Exception;
import org.yaac.server.egql.exception.EGQLException;
import org.yaac.server.egql.processor.ChannelMsgSender;
import org.yaac.server.egql.processor.DatastoreLoader;
import org.yaac.server.egql.processor.ProcessData.ProcessDataRecord;
import org.yaac.server.egql.processor.Processor;
import org.yaac.server.egql.processor.SelectProccesor;
import org.yaac.shared.egql.EGQLConstant;

import com.google.common.collect.ImmutableSet;

/**
 * @author Max Zhu (thebbsky@gmail.com)
 *
 */
public class SelectStatement extends Statement {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	
	private SelectClause selectClause;
	
	private FromClause fromClause;

	private WhereClause whereClause;
	
	private GroupByClause groupByClause;
	
	private HavingClause havingClause;
	
	public SelectClause getSelectClause() {
		return selectClause;
	}

	public void setSelectClause(SelectClause selectClause) {
		this.selectClause = selectClause;
	}

	public FromClause getFromClause() {
		return fromClause;
	}

	public void setFromClause(FromClause fromClause) {
		this.fromClause = fromClause;
	}
	
	public WhereClause getWhereClause() {
		return whereClause;
	}

	public void setWhereClause(WhereClause whereClause) {
		this.whereClause = whereClause;
	}

	public GroupByClause getGroupByClause() {
		return groupByClause;
	}

	public void setGroupByClause(GroupByClause groupByClause) {
		this.groupByClause = groupByClause;
	}

	public HavingClause getHavingClause() {
		return havingClause;
	}

	public void setHavingClause(HavingClause havingClause) {
		this.havingClause = havingClause;
	}

	@Override
	public int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result + ((fromClause == null) ? 0 : fromClause.hashCode());
		result = prime * result + ((groupByClause == null) ? 0 : groupByClause.hashCode());
		result = prime * result + ((havingClause == null) ? 0 : havingClause.hashCode());
		result = prime * result + ((selectClause == null) ? 0 : selectClause.hashCode());
		result = prime * result + ((whereClause == null) ? 0 : whereClause.hashCode());
		return result;
	}

	@Override
	public boolean equals(Object obj) {
		if (this == obj)
			return true;
		if (obj == null)
			return false;
		if (getClass() != obj.getClass())
			return false;
		SelectStatement other = (SelectStatement) obj;
		if (fromClause == null) {
			if (other.fromClause != null)
				return false;
		} else if (!fromClause.equals(other.fromClause))
			return false;
		if (groupByClause == null) {
			if (other.groupByClause != null)
				return false;
		} else if (!groupByClause.equals(other.groupByClause))
			return false;
		if (havingClause == null) {
			if (other.havingClause != null)
				return false;
		} else if (!havingClause.equals(other.havingClause))
			return false;
		if (selectClause == null) {
			if (other.selectClause != null)
				return false;
		} else if (!selectClause.equals(other.selectClause))
			return false;
		if (whereClause == null) {
			if (other.whereClause != null)
				return false;
		} else if (!whereClause.equals(other.whereClause))
			return false;
		return true;
	}

	@Override
	public String toString() {
		return "SelectStatement [selectClause=" + selectClause + ", fromClause=" + fromClause + ", whereClause="
				+ whereClause + ", groupByClause=" + groupByClause + ", havingClause=" + havingClause + "]";
	}
	
	/**
	 * @return
	 */
	public boolean isGroupByQuery() {
		return getGroupByClause() != null;
	}
	
	/**
	 * @return true if there are aggregation functions AND no other non-aggregated field evaluators
	 * 
	 * note that aggregation function doesn't has to be on top level
	 * also, non aggregation functions can be expressions without fields like : 1 + 3
	 */
	public boolean aggregationEvaluatorOnly() { 
		boolean allAgg = true;
		for (Evaluator evaluator : getSelectClause().getItems()) {
			if (!evaluator.nonAggregationProperties().isEmpty()) {
				allAgg = false;
				break;	
			}
		}
		
		return allAgg && !isGroupByQuery();
	}

	/**
	 * @return all aggregators in select caluse or having clause
	 */
	public Collection<AggregationEvaluator> getAllAggregationEvaluators() {
		Set<AggregationEvaluator> result = newHashSet();
		
		for (Evaluator e : selectClause.getItems()) {
			result.addAll(e.aggregationChildren());
		}
		
		if (havingClause != null) {
			result.addAll(havingClause.getConditions().aggregationChildren());
		}

		return result;
	}

	@Override
	public void validate() throws EGQLException {		
		// validate all evaluators in select / where having clause
		for (Evaluator e : selectClause.getItems()) {
			e.validate();
		}
		
		if (whereClause != null) {
			whereClause.getConditions().validate();
		}
		
		if (havingClause != null) {
			havingClause.getConditions().validate();
		}

		Set<AggregationEvaluator> aggregationEvaluators = 
			ImmutableSet.<AggregationEvaluator>builder().addAll(this.getAllAggregationEvaluators()).build();
		
		if (!aggregationEvaluators.isEmpty()) {
			// E001 : not a single-group group function
			{
				// all properties appear in select items MUST be in group by clause as well
				Set<EntityProperty> nonAggregationProperties = newHashSet();
				for (Evaluator e : selectClause.getItems()) {
					 nonAggregationProperties.addAll(e.nonAggregationProperties());
				}
				
				if (!nonAggregationProperties.isEmpty()) {
					if (groupByClause == null) {
						throw new EGQLE001Exception();
					} else {
						nonAggregationProperties.removeAll(groupByClause.items());
						if (! nonAggregationProperties.isEmpty()) {
							// some fields are not in group by function 
							throw new EGQLE001Exception();	
						}
					}
				}	
			}
		}
		
		// E002 : havingClause : not a GROUP BY expression
		{
			if (havingClause != null) {	// having clause is optional
				// all properties appear in having items MUST be in group by clause as well
				Set<EntityProperty> s = havingClause.getConditions().nonAggregationProperties();
				
				if (groupByClause != null) {
					// it is possible to have a having clause without group by
					// eg, select count(home_team) from match having count(home_team) < 10
					s.removeAll(groupByClause.items());	
				}
				
				if (! s.isEmpty()) { 
					throw new EGQLE002Exception();	
				}	
			}
		}
	}

	/**
	 * isRejected if data in evaluation context does not match where clause
	 * 
	 * @param record
	 * @return
	 */
	public boolean rejectedByWhereClause(ProcessDataRecord record) {
		return getWhereClause() != null && !whereClause.evaluate(record);
	}

	public boolean rejectedByHavingClause(ProcessDataRecord record) {
		return havingClause != null && !havingClause.evaluate(record);
	}

	@Override
	public List<Processor> generateProcessors() {
		Processor msgSender = new ChannelMsgSender(EGQLConstant.DEFAULT_MAX_RESULT);
		return newArrayList(datastoreLoader(), selector(), msgSender);
	}

	Processor datastoreLoader() {
		return new DatastoreLoader(
				this.getFromClause().getFromEntities().get(0).getEntityName(), EGQLConstant.DEFAULT_BATCH_SIZE);
	}
	
	Processor selector() {
		return new SelectProccesor(this);
	}
	
	@Override
	public boolean isSimpleStatement() {
		// at the moment always return false, return true if it's a very simple select statement like GQL??
		return false;
	}
}
